package com.unicloud.medical.oss.client;

import com.google.protobuf.ByteString;
import com.unicloud.medical.oss.client.properties.OSSClientProperties;
import com.unicloud.medical.oss.client.util.ResultDTOUtil;
import com.unicloud.medical.oss.common.constant.CommonConstant;
import com.unicloud.medical.oss.common.model.BucketInfo;
import com.unicloud.medical.oss.common.model.DownloadObject;
import com.unicloud.medical.oss.common.model.FileMetaData;
import com.unicloud.medical.oss.common.model.ObjectInfo;
import com.unicloud.medical.oss.common.model.request.*;
import com.unicloud.medical.oss.core.OSSServiceFinder;
import com.unicloud.medical.oss.service.grpc.*;
import com.unisound.medical.common.model.dto.ResultDTO;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
public class GrpcOSSClient extends BaseOSSClient {
    private final OSSServiceGrpc.OSSServiceBlockingStub blockStub;
    private final OSSServiceGrpc.OSSServiceStub stub;

    public GrpcOSSClient(OSSClientProperties ossClientProperties, OSSServiceFinder ossServiceFinder, OSSServiceGrpc.OSSServiceBlockingStub blockStub, OSSServiceGrpc.OSSServiceStub stub) {
        super(ossClientProperties, ossServiceFinder);
        if (blockStub == null || stub == null) {
            log.error("grpc暂不支持,请选择其他连接方式");
        }
        this.blockStub = blockStub;
        this.stub = stub;
    }

    /**
     * 查看存储桶是否存在
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "bucketExistFallback")
    @Override
    public ResultDTO<Boolean> doBucketExist(BucketExistRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.bucketExist(BucketExistGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .build()));
    }

    /**
     * 创建存储桶
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "createBucketFallback")
    @Override
    public ResultDTO<Boolean> doCreateBucket(CreateBucketRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.createBucket(CreateBucketGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .build()));
    }

    /**
     * 创建默认权限存储桶(默认策略支持浏览器直接访问url)
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "createDefaultPolicyBucketFallback")
    @Override
    public ResultDTO<Boolean> doCreateDefaultPolicyBucket(CreateBucketRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.createDefaultPolicyBucket(CreateBucketGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .build()));
    }

    /**
     * 删除存储桶
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "removeBucketFallback")
    @Override
    public ResultDTO<Boolean> doRemoveBucket(DeleteBucketRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.removeBucket(DeleteBucketGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .build()));
    }

    /**
     * 获取全部存储桶
     */
    public ResultDTO<List<BucketInfo>> doGetAllBuckets(CommonRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.getAllBuckets(CommonGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .build()));
    }

    /**
     * 获取当前存储桶所有文件
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "listAllObjectsCurrBucketFallback")
    @Override
    public ResultDTO<List<ObjectInfo>> doListAllObjectsCurrBucket(ListAllObjectsCurrBucketRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.listAllObjectsCurrBucket(ListAllObjectsCurrBucketGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                        .setBucketName(request.getBucketName())
                .build()));
    }

    /**
     * 获取当前存储桶当前文件夹所有文件
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "listObjectsCurrBucketAndFolderFallback")
    @Override
    public ResultDTO<List<ObjectInfo>> doListObjectsCurrBucketAndFolder(ListAllObjectsCurrBucketAndFolderRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.listObjectsCurrBucketAndFolder(ListAllObjectsCurrBucketAndFolderGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                        .setFolder(request.getFolder())
                .build()));
    }

    /**
     * 文件上传
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "uploadObjectsFallback")
    @Override
    public ResultDTO<List<String>> doUploadObjects(InnerUploadObjectsRequest request) {
        final CountDownLatch finishLatch = new CountDownLatch(1);

        ResultDTO<List<String>> result = ResultDTO.fail("GRPC_TIMEOUT", "GRPC超时");

        StreamObserver<UploadObjectsGrpcRequest> requestObserver = stub.uploadObjects(new StreamObserver<GrpcResultDTOV3>() {
            @Override
            public void onNext(GrpcResultDTOV3 o) {
                ResultDTO<List<String>> resultDTO = ResultDTOUtil.toResultDTO(o);
                to(resultDTO, result);
            }

            @Override
            public void onError(Throwable e) {
                finishLatch.countDown();
            }

            @Override
            public void onCompleted() {
                finishLatch.countDown();
            }
        });

        try {
            UploadObjectsGrpcRequest grpcRequest = UploadObjectsGrpcRequest.newBuilder()
                    .setOssName(request.getOssName())
                    .setBucketName(request.getBucketName())
                    .setFolder(request.getFolder())
                    .addAllFiles(request.getFiles().stream()
                            .map(file -> FileMetaDataGrpcRequest.newBuilder()
                                    .setData(ByteString.copyFrom(file.getData()))
                                    .setContentType(file.getContentType())
                                    .setSize(file.getSize())
                                    .setFileName(file.getFileName())
                                    .build())
                            .collect(Collectors.toList()))
                    .build();
            requestObserver.onNext(grpcRequest);

            if (finishLatch.getCount() == 0) {
                // RPC completed or errored before we finished sending.
                // Sending further requests won't error, but they will just be thrown away.
                return result;
            }
        } catch (Throwable e) {
            requestObserver.onError(e);
            return result;
        }

        requestObserver.onCompleted();

        try {
            finishLatch.await(1, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            log.error("doUploadObjects error.", e);
        }

        return result;
    }

    private <T> void to(ResultDTO<T> a, ResultDTO<T> b) {
        b.setCode(a.getCode());
        b.setMessage(a.getMessage());
        b.setSuccess(a.isSuccess());
        b.setData(a.getData());
    }

    /**
     * 文件上传
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "uploadObjectFallback")
    @Override
    public ResultDTO<String> doUploadObject(InnerUploadObjectRequest request) {
        final CountDownLatch finishLatch = new CountDownLatch(1);

        ResultDTO<String> result = ResultDTO.fail("GRPC_TIMEOUT", "GRPC超时");

        StreamObserver<UploadObjectGrpcRequest> requestObserver = stub.uploadObject(new StreamObserver<GrpcResultDTOV2>() {
            @Override
            public void onNext(GrpcResultDTOV2 o) {
                ResultDTO<String> resultDTO = ResultDTOUtil.toResultDTO(o);
                to(resultDTO, result);
            }

            @Override
            public void onError(Throwable e) {
                finishLatch.countDown();
            }

            @Override
            public void onCompleted() {
                finishLatch.countDown();
            }
        });

        try {
            FileMetaData file = request.getFile();
            UploadObjectGrpcRequest grpcRequest = UploadObjectGrpcRequest.newBuilder()
                    .setOssName(request.getOssName())
                    .setBucketName(request.getBucketName())
                    .setObjectName(request.getObjectName())
                    .setFile(FileMetaDataGrpcRequest.newBuilder()
                            .setData(ByteString.copyFrom(file.getData()))
                            .setContentType(file.getContentType())
                            .setSize(file.getSize())
                            .setFileName(file.getFileName())
                            .build())
                    .build();
            requestObserver.onNext(grpcRequest);

            if (finishLatch.getCount() == 0) {
                // RPC completed or errored before we finished sending.
                // Sending further requests won't error, but they will just be thrown away.
                return result;
            }
        } catch (Throwable e) {
            requestObserver.onError(e);
            return result;
        }

        requestObserver.onCompleted();

        try {
            finishLatch.await(1, TimeUnit.MINUTES);
        } catch (Exception e) {
            log.error("doUploadObject error.", e);
        }

        return result;
    }

    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "uploadLocalObjectFallback")
    public ResultDTO<String> uploadLocalObject(UploadLocalObjectRequest request) {
        return null;
    }

    /**
     * 批量本地文件上传
     */
    @Override
    public ResultDTO<List<String>> doUploadLocalObjects(UploadLocalObjectsRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.uploadLocalObjects(UploadLocalObjectsGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .setFolder(request.getFolder())
                        .addAllFileNames(request.getFileNames())
                .build()));
    }

    /**
     * 文件下载
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "downloadObjectFallback")
    @Override
    public ResultDTO<DownloadObject> doDownloadObject(DownloadObjectRequest request) {
        Iterator<GrpcResultDTOV6> grpcResultDTOV6Iterator = blockStub.downloadObject(DownloadObjectGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .setObjectName(request.getObjectName())
                .build());
        while (grpcResultDTOV6Iterator.hasNext()) {
            GrpcResultDTOV6 next = grpcResultDTOV6Iterator.next();
            return ResultDTOUtil.toResultDTO(next);
        }

        return null;
    }

    /**
     * 下载文件夹下所有文件
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "downloadObjectsFallback")
    @Override
    public ResultDTO<DownloadObject> doDownloadObjects(DownloadObjectsRequest request) {
        Iterator<GrpcResultDTOV6> grpcResultDTOV6Iterator = blockStub.downloadObjects(DownloadObjectsGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .setFolder(request.getFolder())
                .setFileName(request.getFileName())
                .build());

        while (grpcResultDTOV6Iterator.hasNext()) {
            GrpcResultDTOV6 next = grpcResultDTOV6Iterator.next();
            return ResultDTOUtil.toResultDTO(next);
        }

        return null;
    }

    /**
     * 根据url地址删除文件
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "removeObjectFallback")
    @Override
    public ResultDTO<String> doRemoveObject(RemoveObjectRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.removeObject(RemoveObjectGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .setUrl(request.getUrl())
                .build()));
    }

    /**
     * 根据url地址列表批量删除文件
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "removeObjectsFallback")
    @Override
    public ResultDTO<List<String>> doRemoveObjects(RemoveObjectsRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.removeObjects(RemoveObjectsGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .addAllUrls(request.getUrls())
                .build()));
    }

    /**
     * 删除当前桶当前文件夹所有文件
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "removeObjectsCurrBucketAndFolderFallback")
    @Override
    public ResultDTO<List<String>> doRemoveObjects(RemoveObjectsCurrBucketAndFolderRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.removeObjectsCurrFolder(RemoveObjectsCurrBucketAndFolderGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .setFolder(request.getFolder())
                .build()));
    }

    /**
     * 文件复制
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "copyObjectFallback")
    @Override
    public ResultDTO<String> doCopyObject(CopyObjectRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.copyObject(CopyObjectGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .setSourceObjectName(request.getSourceObjectName())
                .setTargetObjectName(request.getTargetObjectName())
                .build()));
    }

    /**
     * 已签名地址
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "signedUrlFallback")
    @Override
    public ResultDTO<String> doSignedUrl(SignedUrlRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.signedUrl(SignedUrlGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .setBucketName(request.getBucketName())
                .setObjectName(request.getObjectName())
                .build()));
    }

    /**
     * 批量已签名地址
     */
    @CircuitBreaker(name = CommonConstant.DEFAULT_CIRCUIT_BREAKER_NAME, fallbackMethod = "signedUrlsFallback")
    @Override
    public ResultDTO<List<String>> doSignedUrls(SignedUrlsRequest request) {
        return ResultDTOUtil.toResultDTO(blockStub.signedUrls(SignedUrlsGrpcRequest.newBuilder()
                .setOssName(request.getOssName())
                .addAllObjectNames(request.getObjectNames())
                .build()));
    }

}
